package my.thesis.bolts.aggregators;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import my.thesis.analytics.metrics.ThroughputMetric;
import my.thesis.utilities.JoinFieldsTupleSentiment;
import my.thesis.utilities.JoinLangHashTuple;
import backtype.storm.Config;
import backtype.storm.Constants;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class SameLangHash extends BaseRichBolt
{
	private static final long serialVersionUID = 1L;
	private ConcurrentHashMap<Long, HashMap<String, JoinLangHashTuple>> sameHashes;
	private OutputCollector collector;
	
	//private String lang_1="en";
	
	private String lang_1="de";
	private String lang_2="fr";

	private Boolean receivedSinceLastEmit;
	
	private int secToEmit;
	private Long emitUpToTimeInterval;
	
	transient ThroughputMetric _throughputMetric;
	
	private static Boolean metricsOn=false;
	
	public SameLangHash(int emSec)
	{
		this.secToEmit=emSec;
	}
	
	public void prepare(Map stormConf, TopologyContext context,OutputCollector collector) 
	{
		sameHashes = new ConcurrentHashMap<Long, HashMap<String, JoinLangHashTuple>>();
		this.collector=collector;
		
		this.receivedSinceLastEmit=false;
		
		if(metricsOn)
		{
			_throughputMetric=new ThroughputMetric();
			context.registerMetric("sameHashtagCount", _throughputMetric, 20);
			_throughputMetric.setStartTime();
		}
	}
	
    private HashMap<String, JoinLangHashTuple> getMapForTi(Long timeInterval) 
    {
    	try
    	{
    	  HashMap<String, JoinLangHashTuple> countForInterval = sameHashes.get(timeInterval);
		    
		  if (countForInterval == null) 
		  {
		    	HashMap<String,JoinLangHashTuple> countForIntervalToAdd=new HashMap<String, JoinLangHashTuple>();
		    	sameHashes.put(timeInterval, countForIntervalToAdd);
		    	return sameHashes.get(timeInterval);
		   }
		  else 
		  { 
				return  sameHashes.get(timeInterval);
		  }
    	}
    	catch(Exception ex)
    	{
    		System.out.println(ex.toString());
    		return null;
    	}
	}
	
	public void declareOutputFields(OutputFieldsDeclarer declarer) 
	{
		declarer.declare(new Fields("timeInterval", "countsForInterval"));
	}

	public void execute(Tuple tuple) 
	{
		if (isTickTuple(tuple)) 
		{			
			if(this.receivedSinceLastEmit)
			{				
				emitResults();
			}
			
			this.receivedSinceLastEmit=false;
	    }
		else
		{
			this.receivedSinceLastEmit=true;
			
			if(metricsOn)
			{
				//_throughputMetric.incrTuples();
			}
			
			Long ti = tuple.getLongByField("timeInterval");
			
		    this.emitUpToTimeInterval=ti;
			
			String lang=tuple.getStringByField("lang");
			
			HashMap<String, JoinLangHashTuple> mapForTi = getMapForTi(ti);
			
			String hash=tuple.getStringByField("hash");
			
			//an to tuple einai sth prwth glwssa
			if(lang.equals(lang_1))
			{
				//an se ayto to ti to hash to exei vrei
				if(mapForTi.containsKey(hash))
				{
					//kai to exei vrei se allh glwssa
					if(mapForTi.get(hash).getLan2() != null)
					{
						mapForTi.get(hash).setLan1(lang);
						//System.out.println("JOIN FOUND: " + hash + " LANG1 " + mapForTi.get(hash).getLan1() + "/LANG2: " + mapForTi.get(hash).getLan2());
					}
							
				}
				else 
				{
					//vale oti to vrhkes sthn prwth glwssa
					JoinLangHashTuple m= new JoinLangHashTuple(lang,null);
					mapForTi.put(hash, m);
				}			
			}
			
			//an thelw mono agglika kai mh agglika
			//else
			//an to tuple einai sth deyterh glwssa
			else if(lang.equals(lang_2))
			{
				//an se ayto to ti to hash to exei vrei
				if(mapForTi.containsKey(hash))
				{
					//kai to exei vrei sth prwth glwssa
					if(mapForTi.get(hash).getLan1() != null)
					{
						mapForTi.get(hash).setLan2(lang);
						//System.out.println("JOIN FOUND: " + hash + " LANG1 " + mapForTi.get(hash).getLan1() + "/LANG2: " + mapForTi.get(hash).getLan2());
					}
					
					
				}
				else
				{
					//vale oti to vrhkes sth deyterh glwssa
					JoinLangHashTuple m= new JoinLangHashTuple(null,lang);
					mapForTi.put(hash, m);
				}
			}	
		}	
	}
	
	private void emitResults() 
    {
	   //TEST
	   //System.out.println("EMIT UP TO " + this.emitUpToTimeInterval);
	       
      Set<Long> timeIntervalsAvailable = sameHashes.keySet();
      
      for (Long timeInterval : timeIntervalsAvailable) 
      {
        if (timeInterval < this.emitUpToTimeInterval) 
        {
        	//TEST
        	//System.out.println("Emmiting Hashtags...");
        	
        	HashMap<String, JoinLangHashTuple> hashesForTi = sameHashes.remove(timeInterval);
        	
        	//arraylist is serializable
        	List<String> sameHashesForTi = new ArrayList<String>();
        	for(String h : hashesForTi.keySet() )
        	{
        		if(hashesForTi.get(h).getLan1()!=null && hashesForTi.get(h).getLan2()!=null)
        		{
        			sameHashesForTi.add(h);
        		}
        	}
        	
        	collector.emit(new Values(timeInterval, sameHashesForTi));
        }
      }
    }
	
	private boolean isTickTuple(Tuple tuple)
	{
		 String sourceComponent = tuple.getSourceComponent();
		 String sourceStreamId = tuple.getSourceStreamId();
		 return sourceComponent.equals(Constants.SYSTEM_COMPONENT_ID) 
				 && sourceStreamId.equals(Constants.SYSTEM_TICK_STREAM_ID);
	}
	
	@Override
	public Map<String, Object> getComponentConfiguration() 
	{
	    Config conf = new Config();
	    conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, this.secToEmit);
	    return conf;
	}
}
